package com.shujia.card

import java.text.SimpleDateFormat
import java.util.Date

import com.shujia.util.{CarUtil, SparkStreamTool}
import org.apache.spark.streaming.{Duration, Durations}
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.Jedis

object RealTimeCardFlowByWindow extends SparkStreamTool {
  override def run(args: Array[String]): Unit = {
    /**
      * 1.2 按窗口统计每个卡扣流量 - 统计每个卡扣最近5分钟车流量，每隔1分钟统计一次
      *
      */


    val carsDS: DStream[CarUtil.Car] = CarUtil.loadKafkaCar(ssc, "RealTimeCardFlowByWindow")


    val kvDS: DStream[(Long, Int)] = carsDS.map(car => {
      (car.card, 1)
    })

    val countDS: DStream[(Long, Int)] = kvDS.reduceByKeyAndWindow(
      (x: Int, y: Int) => x + y,
      (i: Int, j: Int) => i - j,
      Durations.minutes(5),
      Durations.minutes(1)
    )

    //将数据保存到redis中

    countDS.foreachRDD(rdd => {
      rdd.foreachPartition(iter => {

        //1、简历redis的链接
        val jedis = new Jedis("master", 6379)

        //获取窗口介绍时间
        val date = new Date
        val format = new SimpleDateFormat("yyyyMMddHHmmss")
        val time: String = format.format(date)


        iter.foreach {
          case (card: Long, num: Int) =>
            //构建key
            val key: String = "RealTimeCardFlowByWindow:" + card + ":" + time

            jedis.set(key, num.toString)
            jedis.expire(key, 36000)

        }

        jedis.close()
      })
    })



  }
}
